<?php
/*
 +----------------------------------------------------------------------
 + Title        : Amqp 控制器
 + Author       : Randy_chen
 + Version      : V1.0.0.1
 + Initial-Time : 2020/9/2 22:00
 + Last-time    : 2020/9/2 22:00+ 86187
 + Desc         : Amqp
 +----------------------------------------------------------------------
*/

namespace randy\queue\job;

use randy\queue\Job;
use randy\queue\connector\Amqp as AmqpQueue;

class Amqp extends Job {
	/**
	 * The redis queue instance.
	 *
	 * @var AmqpQueue
	 */
	public $amqp;

	/**
	 * The database job payload.
	 *
	 * @var Object
	 */
	protected $job;

	public function __construct(AmqpQueue $amqp, $job, $queue) {
		$this->job = $job;
		$this->queue = $queue;
		$this->amqp = $amqp;
	}

	/**
	 * 执行队列
	 * Fire the job.
	 *
	 * @return bool
	 */
	public function fire() : bool {
		return $this->resolveAndFire(json_decode($this->getRawBody(), true));
	}

	/**
	 * Get the number of times the job has been attempted.
	 *
	 * @return int
	 */
	public function attempts() {
		return json_decode($this->job, true)['attempts'];
	}

	/**
	 * Get the number of times the job has been attempted.
	 *
	 * @return int
	 */
	public function msg_id() {
		return json_decode($this->job, true)['msg_id'];
	}

	/**
	 * Get the number of times the job has been attempted.
	 *
	 * @return int
	 */
	public function data() {
		return json_decode($this->job, true)['data'];
	}

	/**
	 * Get the raw body string for the job.
	 *
	 * @return string
	 */
	public function getRawBody() {
		return $this->job;
	}

	/**
	 * 删除任务
	 *
	 * @return void
	 */
	public function delete() {
		parent::delete();

		$this->amqp->deleteReserved($this->queue, $this->job);
	}

	/**
	 * 重新发布任务
	 *
	 * @param int $delay
	 *
	 * @return void
	 * @throws \Exception
	 */
	public function release($delay = 0) {
		parent::release($delay);

		$this->delete();
		//发起重试
		$this->amqp->release($this->queue, $this->job, $delay, $this->attempts() + 1);
	}

	/**
	 *
	 * @param $job
	 *
	 * @throws \Exception
	 * @author : Randy_chen
	 * @Date   : 2020/9/3
	 * @Time   : 15:52
	 */
	public function ack(Amqp $job) {
		$this->amqp->ackMsg($job->msg_id());
	}
}